package CheckPoint;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class CheckPointTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 检查点配置,第一个参数是参数是保存周期
         */
        env.enableCheckpointing(300);
        //高级选项
        /**
         * 一致性模式的选择，exactly_once精确一致性，
         * last_lease_once
         */
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //超时时间
        env.getCheckpointConfig().setCheckpointTimeout(1000);
        //最大的同时进行的checkpoints，因为checkpoint保存火进行处理需要耗费时间，所以可能导致一个checkpoint正在保存时，另一个checkpoint却已经启动了
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //俩个checkpoints间的最小间歇时间(前一个checkpoint保存完成到后一个checkpoint开始的这段时间)
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100l);
        //倾向于使用检查点来进行恢复，默认是false，是使用最近的一个savepoint
        env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
        //允许checkpoint失败的次数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);

        /**
         * 重启策略的配置
         *
         */
        //固定延迟重启，俩个参数一个是重启的次数，一个参数是重启时间的间隔
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,1000l));
        //失败率重启,第一个参数是失败的次数超过三次就算失败，第二个参数十分钟内一直重启，第三个参数每次重启的间隔是一分钟
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10),Time.minutes(1)));
    }
}
